package com.wdl.ad.app

import com.wdl.ad.bean.{AdClickEvent, AdCountByProvince, BlackListWarning}
import com.wdl.ad.function.{AdCountAgg, AdCountResult}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

import java.net.URL


object AdApp {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/AdClickLog.csv")
    val path: String = resource.getPath

    val dataStream: DataStream[String] = env.readTextFile(path)

    val adLogStream: DataStream[AdClickEvent] = dataStream.map(data => {
      val dataArr: Array[String] = data.split(",")
      AdClickEvent(dataArr(0).toLong, dataArr(1).toLong, dataArr(2), dataArr(3), dataArr(4).toLong)
    }).assignAscendingTimestamps(_.timestamp * 1000L)

    val filterStream: DataStream[AdClickEvent] = adLogStream
      /** 按照用户和广告的 id 分组 */
      .keyBy(d => (d.userId, d.adId))
      .process(new FilterBlackList(100L))

    /** 按照 province 分组开窗聚合统计 */
    val adCountStream: DataStream[AdCountByProvince] = filterStream
      .keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new AdCountAgg, new AdCountResult)

    filterStream.getSideOutput(new OutputTag[BlackListWarning]("blackList")).print("output")

    adCountStream.print("adCountStream")


    env.execute("ad job")

  }
}
